#include "common.h"
#include "config.h"
#include "cuckoo.h"
#include "arraylist.h"

#define TEST_NZ(x) do { if ( (x)) die("error: " #x " failed (returned non-zero)." ); } while (0)
#define TEST_Z(x)  do { if (!(x)) die("error: " #x " failed (returned zero/null)."); } while (0)

typedef struct slot_info
{
    int type; //0=client,1=datanodeasserver，2=datanodeasclient
    int nodeid;
    int slot;
}slot_info;

DataNode* datanodes=NULL;
int nodeid;
struct Context *s_ctx = NULL;
Connection** datanodes_conns_as_server=NULL;
Connection** datanodes_conns_as_client=NULL;
int** datanodes_conns_as_server_send_bitmap=NULL;
int** datanodes_conns_as_client_send_bitmap=NULL;
struct rdma_event_channel *event_channel = NULL;
int flag=0;
long long id=0;
cuckoo* update_ctx_map=NULL;
cuckoo* update_entity_map=NULL;
char update_block[BLK_SIZE];
int parity[BLK_SIZE];
pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond=PTHREAD_COND_INITIALIZER;

//================================================================================
void connect2datanodes();
void* thread4connect2datanodes();
int on_connect_request(struct rdma_cm_id *id);
int on_connection(struct rdma_cm_id *id);
int on_disconnect(struct rdma_cm_id *id);
void destroy_connection(void *context);
int on_addr_resolved(struct rdma_cm_id *id);
int on_route_resolved(struct rdma_cm_id *id);
int on_event(struct rdma_cm_event *event);
void build_context(struct ibv_context *verbs);
void build_qp_attr(struct ibv_qp_init_attr *qp_attr);
void register_memory(Connection *conn);
void build_params(struct rdma_conn_param *params);
void post_receives(struct Connection *conn);
void* poll_cq(void *ctx);
void on_completion(struct ibv_wc *wc);
void send_WriteResponse(WriteResponse* response,Connection* conn);
void send_ReadResponse(ReadResponse* response,Connection* conn);
void send_UpdateEntity(UpdateEntity* entity,Client_Connection* client_conn,int paritynodeid);
void send_UpdateCheck(UpdateCheck* check,Connection* conn);
void process_client_request(Connection* conn);
void process_datanode_updateEntity(Connection* conn);
void process_datanode_updateCheck(Connection* conn);

int main(int argc, char const *argv[])
{
    update_ctx_map = cuckoo_init(100);
    update_entity_map = cuckoo_init(100);
    datanodes = read_config_Node(NODE_CONFIG);
	  nodeid = current_node(datanodes);
    datanodes_conns_as_server = (Connection**)malloc(sizeof(Connection*)*DATANODE_NUM);
    datanodes_conns_as_client = (Connection**)malloc(sizeof(Connection*)*DATANODE_NUM);
    datanodes_conns_as_server_send_bitmap = (int**)malloc(sizeof(int*)*DATANODE_NUM);
    datanodes_conns_as_client_send_bitmap = (int**)malloc(sizeof(int*)*DATANODE_NUM);

    for (int i = 0; i < DATANODE_NUM; ++i)
    {
        datanodes_conns_as_server_send_bitmap[i] = (int*)malloc(sizeof(int)*PARALLEL);
        datanodes_conns_as_client_send_bitmap[i] = (int*)malloc(sizeof(int)*PARALLEL);
    }

    for (int i = 0; i < DATANODE_NUM; ++i)
    {
       for (int j = 0; j < PARALLEL; ++j)
       {
         datanodes_conns_as_client_send_bitmap[i][j] = 0;
         datanodes_conns_as_server_send_bitmap[i][j] = 0;
       }
    }


    //创建连接线程
    pthread_t connect2datanodes_threadid;
    pthread_create(&connect2datanodes_threadid,NULL,thread4connect2datanodes,NULL);

	  struct sockaddr_in addr;
	  struct rdma_cm_event *event = NULL;
  	struct rdma_cm_id *listener_id = NULL;	
  	uint16_t port = RDMA_PORT;
  	memset(&addr, 0, sizeof(addr));
  	addr.sin_family = AF_INET;
  	addr.sin_port = htons(port);
  	addr.sin_addr.s_addr = htonl(INADDR_ANY);

  	TEST_Z(event_channel=rdma_create_event_channel());
  	TEST_NZ(rdma_create_id(event_channel,&listener_id,NULL,RDMA_PS_TCP));
  	TEST_NZ(rdma_bind_addr(listener_id,(struct sockaddr *)&addr));
  	TEST_NZ(rdma_listen(listener_id,10));//backlog=10

  	printf("datanode[%s:%d] is listening now,\n",datanodes[nodeid].rdma_ip,port);
  	printf("current_node id is %d\n",nodeid);

  	while (rdma_get_cm_event(event_channel, &event) == 0) {
	    struct rdma_cm_event event_copy;
	    memcpy(&event_copy, event, sizeof(*event));
	    rdma_ack_cm_event(event);
	    if (on_event(&event_copy))
	    	break;
  	}

    rdma_destroy_id(listener_id);
    rdma_destroy_event_channel(event_channel);

    pthread_cancel(connect2datanodes_threadid);
	  return 0;
}

//连接其他datanodes 用于rdma-write
void connect2datanodes(){

  for (int i = 0; i < DATANODE_NUM; ++i)
  {
    flag=0;
    if (i!=nodeid)
    {
      char* target_ip=datanodes[i].rdma_ip;
      //与目标datanode建立连接，并且将conn存入datanodes_conns_as_server
      //note:不可与自己连接，故数组中有nodeid坐标元素为NULL
      struct rdma_cm_id *conn_id= NULL;
      // struct sockaddr_in *dst_addr = (struct sockaddr_in*)malloc(sizeof(struct sockaddr_in));
      // dst_addr->sin_family=AF_INET;
      // dst_addr->sin_port = htons(RDMA_PORT);
      // dst_addr->sin_addr.s_addr = inet_addr(target_ip);
      struct addrinfo *addr;
      char port[20];
      sprintf(port,"%d",RDMA_PORT);
      TEST_NZ(getaddrinfo(target_ip, port,NULL,&addr));
      TEST_NZ(rdma_create_id(event_channel, &conn_id, NULL, RDMA_PS_TCP));
      TEST_NZ(rdma_resolve_addr(conn_id, NULL, addr->ai_addr, TIMEOUT_IN_MS));
      printf("%s\n", target_ip);
      while(flag==0){};
    }
  }
}

void* thread4connect2datanodes(void* arg){
    sleep(10);
    connect2datanodes();
}

int on_connect_request(struct rdma_cm_id *id)
{
  printf("received connection request.\n");

  //根据id判断来自client连接或是来自datanode连接
  struct sockaddr* addr =  rdma_get_peer_addr(id);
  int peer_id = get_nodeid(addr,datanodes);
  printf("the connection request comes from:%d\n", peer_id);

  //上下文,保护域,完成队列
  build_context(id->verbs);
  //创建QP列队,设置QP参数
  struct ibv_qp_init_attr qp_attr;
  build_qp_attr(&qp_attr);
  //qp连接
  TEST_NZ(rdma_create_qp(id, s_ctx->pd, &qp_attr));

  Connection* conn = (Connection*)malloc(sizeof(Connection));
  if (peer_id==-1)//client
  {
    conn->type = 0;
    Client_Connection* client_conn = (Client_Connection*)malloc(sizeof(Client_Connection));
    conn->conn_info = (void*)client_conn;

    // for (int i = 0; i < DATANODE_NUM; ++i)
    // {
    //   if (i!=nodeid)
    //   {
    //     char* target_ip=datanodes[i].rdma_ip;
    //     struct rdma_cm_id *conn_id= NULL;
    //     struct addrinfo *addr;
    //     char port[20];
    //     sprintf(port,"%d",RDMA_PORT);
    //     TEST_NZ(getaddrinfo(target_ip, port,NULL,&addr));
    //     TEST_NZ(rdma_create_id(event_channel, &conn_id, NULL, RDMA_PS_TCP));
    //     TEST_NZ(rdma_resolve_addr(conn_id, NULL, addr->ai_addr, TIMEOUT_IN_MS));

    //     client_conn->conn_ids[i] = conn_id;
    //     printf("%s\n", target_ip);
    //   }
    // }

  }else{//datanode
    conn->type = 1;
    DataNode_Connection* datanode_conn = (DataNode_Connection*)malloc(sizeof(DataNode_Connection));
    conn->conn_info = (void*)datanode_conn;
    datanodes_conns_as_server[peer_id] = conn;//将连接的datanode连接信息按照nodeid保存
  }
  conn->qp = id->qp;
  conn->id = id;
  conn->len=0;
  id->context = conn;
  //注册内存区域
  register_memory(conn);
  
  struct rdma_conn_param cm_params;
  //创建连接参数  
  build_params(&cm_params);
  //接受来自client或者datanode的连接
  TEST_NZ(rdma_accept(id, &cm_params));

  //发送一个receive消息
  post_receives(conn);
  return 0;
}


int on_connection(struct rdma_cm_id *id)
{
  Connection* conn = (Connection*)id->context;
  if (conn->type==0)
  {
    printf("accept client connection successfully\n");
  }else if (conn->type==1)
  {
    printf("accept datanode connection successfully\n");
  }else if(conn->type==2){
    flag=1;
    printf("connect to datanode successfully\n");
  }else{
    printf("error connection type\n");
    return -1;
  }
  return 0;
}

int on_disconnect(struct rdma_cm_id *id)
{
  printf("peer disconnected.\n");
  destroy_connection(id->context); 
  return 0;
}

void destroy_connection(void *context)
{
  struct Connection *conn = (struct Connection *)context;
  rdma_destroy_qp(conn->id);

  if (conn->type==0)
  {
      Client_Connection* client_conn = (Client_Connection*)conn->conn_info;
      ibv_dereg_mr(client_conn->send_mr);
      ibv_dereg_mr(client_conn->recv_mr);
      free(client_conn->recv_region);
      free(client_conn->send_region); 
  }else if (conn->type==1||conn->type==2)
  {
      DataNode_Connection* datanode_conn = (DataNode_Connection*)conn->conn_info;
      ibv_dereg_mr(datanode_conn->send_mr);
      ibv_dereg_mr(datanode_conn->recv_mr);
      free(datanode_conn->recv_region);
      free(datanode_conn->send_region);
  }else{
      printf("unknown conn type\n");
  }
  rdma_destroy_id(conn->id);
  free(conn);
}

int on_addr_resolved(struct rdma_cm_id *id)
{
  printf("start address resolved.\n");
  struct sockaddr* addr =  rdma_get_peer_addr(id);

  int peer_id = get_nodeid(addr,datanodes);

  printf("on_addr_resolved:%d\n",peer_id);

  struct ibv_qp_init_attr qp_attr;
  build_context(id->verbs);
  build_qp_attr(&qp_attr);
  TEST_NZ(rdma_create_qp(id, s_ctx->pd, &qp_attr));

  Connection* conn = (Connection*)malloc(sizeof(Connection));
  conn->type=2;
  conn->qp = id->qp;
  conn->id = id;
  conn->len=0;
  id->context = conn;
  datanodes_conns_as_client[peer_id] = conn;
  DataNode_Connection* datanode_conn = (DataNode_Connection*)malloc(sizeof(DataNode_Connection));
  conn->conn_info = (void*)datanode_conn;

  register_memory(conn);
  TEST_NZ(rdma_resolve_route(id, TIMEOUT_IN_MS));
  printf("end address resolved.\n");
  post_receives(conn);
  return 0;
}

int on_route_resolved(struct rdma_cm_id *id)
{
  struct rdma_conn_param cm_params;
  printf("route resolved.\n");
  build_params(&cm_params);
  TEST_NZ(rdma_connect(id, &cm_params));
  return 0;
}

int on_event(struct rdma_cm_event *event)
{
  int r = 0;
  printf("%s\n",rdma_event_str(event->event));
  if (event->event == RDMA_CM_EVENT_ADDR_RESOLVED)
    r = on_addr_resolved(event->id);
  else if (event->event == RDMA_CM_EVENT_ROUTE_RESOLVED)
    r = on_route_resolved(event->id);
  else if (event->event == RDMA_CM_EVENT_CONNECT_REQUEST)
    r = on_connect_request(event->id);
  else if (event->event == RDMA_CM_EVENT_ESTABLISHED)
    r = on_connection(event->id);
  else if (event->event == RDMA_CM_EVENT_DISCONNECTED)
    r = on_disconnect(event->id);
  else{
    die("on_event: unknown event.");
    //printf("unknown event.\n");
  }
  return r;
}


void build_context(struct ibv_context *verbs)
{
  if (s_ctx) {
    if (s_ctx->ctx != verbs)
      die("cannot handle events in more than one context.");
    return;
  }

  s_ctx = (struct Context *)malloc(sizeof(struct Context));

  s_ctx->ctx = verbs;

  TEST_Z(s_ctx->pd = ibv_alloc_pd(s_ctx->ctx));
  TEST_Z(s_ctx->comp_channel = ibv_create_comp_channel(s_ctx->ctx));
  TEST_Z(s_ctx->cq = ibv_create_cq(s_ctx->ctx, 32, NULL, s_ctx->comp_channel, 0)); /* cqe=10 is arbitrary */
  TEST_NZ(ibv_req_notify_cq(s_ctx->cq, 0));

  TEST_NZ(pthread_create(&s_ctx->cq_poller_thread, NULL, poll_cq, NULL));
}

void build_qp_attr(struct ibv_qp_init_attr *qp_attr)
{
  memset(qp_attr, 0, sizeof(*qp_attr));

  qp_attr->send_cq = s_ctx->cq;
  qp_attr->recv_cq = s_ctx->cq;
  qp_attr->qp_type = IBV_QPT_RC;

  qp_attr->cap.max_send_wr = 10;
  qp_attr->cap.max_recv_wr = 10;
  qp_attr->cap.max_send_sge = 1;
  qp_attr->cap.max_recv_sge = 1;
}

void register_memory(Connection *conn)
{
  if (conn->type==0)//client
  {
    Client_Connection* client_conn = (Client_Connection*)conn->conn_info;
    client_conn->send_region = (char*)malloc(sizeof(ReadResponse));
    client_conn->recv_region = (char*)malloc(sizeof(WriteRequest));

    TEST_Z(client_conn->send_mr = ibv_reg_mr(
          s_ctx->pd,
          client_conn->send_region,
          sizeof(ReadResponse),
          IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE));

    TEST_Z(client_conn->recv_mr = ibv_reg_mr(
          s_ctx->pd,
          client_conn->recv_region,
          sizeof(WriteRequest),
          IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE));

  }else if (conn->type==1)
  {//datanode
      DataNode_Connection* datanode_conn = (DataNode_Connection*)conn->conn_info;
      datanode_conn->send_region = (char*)malloc(sizeof(UpdateCheck)*PARALLEL);
      datanode_conn->recv_region = (char*)malloc(sizeof(UpdateEntity));

      TEST_Z(datanode_conn->send_mr = ibv_reg_mr(
            s_ctx->pd,
            datanode_conn->send_region,
            sizeof(UpdateCheck)*PARALLEL,
            IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE));

      TEST_Z(datanode_conn->recv_mr = ibv_reg_mr(
            s_ctx->pd,
            datanode_conn->recv_region,
            sizeof(UpdateEntity),
            IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE));
  }else if(conn->type==2){
      DataNode_Connection* datanode_conn = (DataNode_Connection*)conn->conn_info;
      datanode_conn->send_region = (char*)malloc(sizeof(UpdateEntity)*PARALLEL);
      datanode_conn->recv_region = (char*)malloc(sizeof(UpdateCheck));

      TEST_Z(datanode_conn->send_mr = ibv_reg_mr(
            s_ctx->pd,
            datanode_conn->send_region,
            sizeof(UpdateEntity)*PARALLEL,
            IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE));

      TEST_Z(datanode_conn->recv_mr = ibv_reg_mr(
            s_ctx->pd,
            datanode_conn->recv_region,
            sizeof(UpdateCheck),
            IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE));
  }
  else{
    printf("error type in register_memory");
    return;
  }
}


void build_params(struct rdma_conn_param *params)
{
  memset(params, 0, sizeof(*params));
  params->initiator_depth = params->responder_resources = 1;
  params->rnr_retry_count = 7; /* infinite retry */
}


void post_receives(struct Connection *conn)
{
    struct ibv_recv_wr wr, *bad_wr = NULL;
    struct ibv_sge sge;
    //准备工作请求
    wr.wr_id = (uintptr_t)conn;
    wr.next = NULL;
    wr.sg_list = &sge;
    wr.num_sge = 1;
    
    if (conn->type==0){
      Client_Connection* client_conn = (Client_Connection*)conn->conn_info;
      sge.addr = (uintptr_t)client_conn->recv_region;
      sge.length = sizeof(struct WriteRequest);
      sge.lkey = client_conn->recv_mr->lkey;
    }else if(conn->type==1){
      DataNode_Connection* datanode_conn = (DataNode_Connection*)conn->conn_info;
      sge.addr = (uintptr_t)datanode_conn->recv_region;
      sge.length = sizeof(struct UpdateEntity);
      sge.lkey = datanode_conn->recv_mr->lkey;
    }else if(conn->type==2){
      DataNode_Connection* datanode_conn = (DataNode_Connection*)conn->conn_info;
      sge.addr = (uintptr_t)datanode_conn->recv_region;
      sge.length = sizeof(struct UpdateCheck);
      sge.lkey = datanode_conn->recv_mr->lkey;  
    }else{
      die("error connection type");
    }
    TEST_NZ(ibv_post_recv(conn->qp, &wr, &bad_wr));
}

void *poll_cq(void *ctx)
{
  struct ibv_cq *cq;
  struct ibv_wc wc;

  while (1) {
    TEST_NZ(ibv_get_cq_event(s_ctx->comp_channel, &cq, &ctx));
    ibv_ack_cq_events(cq, 1);
    TEST_NZ(ibv_req_notify_cq(cq, 0));

    while (ibv_poll_cq(cq, 1, &wc))
      on_completion(&wc);
  }

  return NULL;
}

//request的处理逻辑
void on_completion(struct ibv_wc *wc)
{
    printf("start sloving the wc\n");
    if (wc->status != IBV_WC_SUCCESS)
      die("on_completion: status is not IBV_WC_SUCCESS.");
    if (wc->opcode == IBV_WC_RECV){
      //三种recv可能，分别处理
      Connection* conn = (struct Connection *)(uintptr_t)wc->wr_id;
      if (conn->type==0){
        printf("start process_client_request\n");      
        process_client_request(conn);
      }else if(conn->type==1){
        printf("start process_datanode_updateEntity\n");
        process_datanode_updateEntity(conn);
      }else if(conn->type==2){
        printf("start process_datanode_updateCheck\n");
        process_datanode_updateCheck(conn);
      }else{
        die("error connection type.");
      }
       post_receives(conn);//放在process之后，放置内容覆盖
    }
    else if (wc->opcode == IBV_WC_SEND){
      slot_info* info = (slot_info *)(uintptr_t)wc->wr_id;
      if (info->type==0)
      {
        printf("send WriteResponse completed successfully.\n");
      }else if (info->type==1)
      {
        datanodes_conns_as_server_send_bitmap[info->nodeid][info->slot]=0;
        printf("send UpdateCheck successfully.\n");
      }else if (info->type==2)
      {
        datanodes_conns_as_client_send_bitmap[info->nodeid][info->slot]=0;
        printf("send UpdateEntity successfully.\n");
      }else
      {
        die("error connection type.");
      }
      free(info);
      // if (conn->type==0){      
      //   printf("send WriteResponse completed successfully.\n");
      // }else if(conn->type==1){
      //   printf("send UpdateCheck successfully.\n");
      // }else if(conn->type==2){
      //   printf("send UpdateEntity successfully.\n");
      // }else{
      //   die("error connection type.");
      // }
    }else
      die("on_completion: completion isn't a send or a receive.");
}

void process_datanode_updateCheck(Connection* conn){
    DataNode_Connection* datanode_conn = (DataNode_Connection*)conn->conn_info;
    UpdateCheck* check = (UpdateCheck*)malloc(sizeof(UpdateCheck));
    memcpy(check,datanode_conn->recv_region,sizeof(UpdateCheck));


    //先判断res为ready还是ok

    char* key = convert_id_to_string(check->id);
    UpdateContext* update_ctx =  (UpdateContext*)cuckoo_get(update_ctx_map,key);

    if(check->res==1){
      update_ctx->ready--;

      if (update_ctx->ready==0)
      {
        UpdateEntity* entity = (UpdateEntity*)malloc(sizeof(UpdateEntity));
        entity->id = update_ctx->id;
        entity->type = 1;
        for (int i = 0; i < R; ++i)
        {
          send_UpdateEntity(entity,(Client_Connection*)update_ctx->conn->conn_info,update_ctx->request->paritynodeid[i]);
        }
        free(entity);
      }
    }

    if (check->res==2)
    {
      update_ctx->status--;
      if (update_ctx->status==0)
      {
        memcpy(update_block,update_ctx->request->newdata,BLK_SIZE);
        WriteResponse* response = (WriteResponse*)malloc(sizeof(WriteResponse));
        response->type=1;
        response->responseid=update_ctx->request->requestid;
        response->res=1;
        send_WriteResponse(response,update_ctx->conn);
        free(response);
        cuckoo_remove(update_ctx_map,key);
        free(update_ctx->request);
        free(update_ctx);
      } 
    }
    free(check);
    free(key);
}

void process_datanode_updateEntity(Connection* conn){
    DataNode_Connection* datanode_conn = (DataNode_Connection*)conn->conn_info;
    UpdateEntity* entity = (UpdateEntity*)malloc(sizeof(UpdateEntity));
    memcpy(entity,datanode_conn->recv_region,sizeof(UpdateEntity));

    if (entity->type==0)
    {
      char* key = convert_id_to_string(entity->id);
      cuckoo_insert(update_entity_map,key,(void*)entity);
      
      UpdateCheck* check = (UpdateCheck*)malloc(sizeof(UpdateCheck));
      check->id = entity->id;
      check->res = 1;
      send_UpdateCheck(check,conn);

      free(key);
      free(check);
    }

    if (entity->type==1)
    {
      char* key = convert_id_to_string(entity->id);
      UpdateEntity* entity2 = (UpdateEntity*)cuckoo_get(update_entity_map,key);
    
      for (int i = 0; i < BLK_SIZE; ++i) parity[i] = parity[i]+entity2->newdata[i];
      UpdateCheck* check = (UpdateCheck*)malloc(sizeof(UpdateCheck));
      check->id = entity2->id;
      check->res = 2;
      send_UpdateCheck(check,conn);

      cuckoo_remove(update_entity_map,key);
      free(key);
      free(entity2);
      free(entity);
      free(check);
    }
}

slot_info* lookforSlot2SendUpdateCheck(Connection* conn){
    slot_info* info = (slot_info*)malloc(sizeof(slot_info));
    info->type=1;
    for (int i = 0; i < DATANODE_NUM; ++i)
    {
      if (conn==datanodes_conns_as_server[i])
      {
        info->nodeid=i;
        for (int j = 0; j < PARALLEL; ++j)
        {
          if (datanodes_conns_as_server_send_bitmap[i][j]==0)
          {
            datanodes_conns_as_server_send_bitmap[i][j]=1;
            info->slot=j;
            return info; 
          }
        }
      }
    }
    info->slot=-1;
    return info;
}

slot_info* lookforSlot2SendUpdateEntity(int paritynodeid){
    slot_info* info = (slot_info*)malloc(sizeof(slot_info));
    info->type=2;
    info->nodeid=paritynodeid;
    for (int i = 0; i < PARALLEL; ++i)
    {
      if (datanodes_conns_as_client_send_bitmap[paritynodeid][i]==0)
      {
        datanodes_conns_as_client_send_bitmap[paritynodeid][i]=1;
        info->slot=i;
        return info;
      }
    }
    info->slot=-1;
    return info;
}

void send_UpdateCheck(UpdateCheck* check,Connection* conn){
    DataNode_Connection* datanode_conn = (DataNode_Connection*)conn->conn_info;

    struct ibv_send_wr wr, *bad_wr = NULL;
    struct ibv_sge sge;
    memset(&wr, 0, sizeof(wr));

    slot_info* info = lookforSlot2SendUpdateCheck(conn);
    if (info->slot==-1)
      die("slot==-1,PARALLEL is small");

    memcpy(datanode_conn->send_region+info->slot*sizeof(UpdateCheck),check,sizeof(UpdateCheck));

    wr.wr_id = (uintptr_t)info;
    wr.opcode = IBV_WR_SEND;
    wr.sg_list = &sge;
    wr.num_sge = 1;
    wr.send_flags = IBV_SEND_SIGNALED;

    sge.addr = (uintptr_t)datanode_conn->send_region+info->slot*sizeof(UpdateCheck);
    sge.length = sizeof(UpdateCheck);
    sge.lkey = datanode_conn->send_mr->lkey;

    TEST_NZ(ibv_post_send(conn->qp, &wr, &bad_wr));
}

void process_client_request(Connection* conn){
    Client_Connection* client_conn = (Client_Connection*)conn->conn_info;

    int trace_type=0;
    memcpy(&trace_type,client_conn->recv_region,sizeof(int));

    if (trace_type==0)
    {
        ReadRequest* request = (ReadRequest*)malloc(sizeof(ReadRequest));
        memcpy(request,client_conn->recv_region,sizeof(ReadRequest));
        //解析ReadRequest，根据blockid找到数据，发送ReadResponse
        ReadResponse* response = (ReadResponse*)malloc(sizeof(ReadResponse));
        response->type=0;
        response->responseid = request->requestid;
        response->res = 1;
        gene_radm_buff(response->newdata,BLK_SIZE);
        send_ReadResponse(response,conn);
        free(response);
        free(request);
    }else if (trace_type==1)
    {
        WriteRequest* request = (WriteRequest*)malloc(sizeof(WriteRequest));
        memcpy(request,client_conn->recv_region,sizeof(WriteRequest));
        //1.解析接收到的writerequest，将冗余同步情况，requestid，更新的数据块信息等封装存储至map中
        UpdateContext* update_ctx = (UpdateContext*)malloc(sizeof(UpdateContext));
        update_ctx->request=request;

        update_ctx->id=getcurrent_time()*10+id++;//用时间代替

        //update_ctx->requestid = request->requestid;
        update_ctx->conn=conn;
        update_ctx->status=R;
        update_ctx->ready=R;
        //update_ctx->blockid=request->blockid;
        //memcpy(update_ctx->newdata,request->newdata,BLK_SIZE);

        char* key = convert_id_to_string(update_ctx->id);
        cuckoo_insert(update_ctx_map,key,(void*)update_ctx);
        free(key);

        //2.发送同步冗余分块的rdma-send请求
        UpdateEntity* entity = (UpdateEntity*)malloc(sizeof(UpdateEntity));
        entity->id = update_ctx->id;
        //memcpy(entity->newdata,request->newdata,BLK_SIZE);
        for (int i = 0; i < BLK_SIZE; ++i) entity->newdata[i] = request->newdata[i]-update_block[i];

        for (int i = 0; i < R; ++i)
        {
          entity->type=0;
          entity->blockid = request->blockid;
          entity->parityid = request->parityid[i];
          entity->checksum = CHECK_SUM;    
          send_UpdateEntity(entity,client_conn,request->paritynodeid[i]);
        }
        //3.等待RDMA-recv完成，根据id拿到conn，更新数据块然后，发送send给client
        free(entity);
    }else
    {
        printf("error trace_type:%d.\n",trace_type);
    }

    
}

void send_UpdateEntity(UpdateEntity* entity,Client_Connection* client_conn,int paritynodeid)
{
    Connection* conn = datanodes_conns_as_client[paritynodeid];
    //rdma-write
    DataNode_Connection* datanode_conn = (DataNode_Connection*)conn->conn_info;

    struct ibv_send_wr wr, *bad_wr = NULL;
    struct ibv_sge sge;
    memset(&wr, 0, sizeof(wr));
    
   slot_info* info = lookforSlot2SendUpdateEntity(paritynodeid);
   if (info->slot==-1)
      die("slot==-1,PARALLEL is small");

    memcpy(datanode_conn->send_region+info->slot*sizeof(UpdateEntity),entity,sizeof(UpdateEntity));
    
    wr.wr_id = (uintptr_t)info;
    wr.opcode = IBV_WR_SEND;
    wr.sg_list = &sge;
    wr.num_sge = 1;
    wr.send_flags = IBV_SEND_SIGNALED;

    sge.addr = (uintptr_t)datanode_conn->send_region+sizeof(UpdateEntity)*info->slot;
    sge.length = sizeof(UpdateEntity);
    sge.lkey = datanode_conn->send_mr->lkey;

    TEST_NZ(ibv_post_send(conn->qp, &wr, &bad_wr));

}

void send_WriteResponse(WriteResponse* response,Connection* conn){
    Client_Connection* client_conn = (Client_Connection*)conn->conn_info;

    struct ibv_send_wr wr, *bad_wr = NULL;
    struct ibv_sge sge;
    memset(&wr, 0, sizeof(wr));

    memcpy(client_conn->send_region,response,sizeof(WriteResponse));

    slot_info* info = (slot_info*)malloc(sizeof(slot_info));
    info->type=0;
    wr.wr_id = (uintptr_t)info;

    wr.opcode = IBV_WR_SEND;
    wr.sg_list = &sge;
    wr.num_sge = 1;
    wr.send_flags = IBV_SEND_SIGNALED;

    sge.addr = (uintptr_t)client_conn->send_region;
    sge.length = sizeof(WriteResponse);
    sge.lkey = client_conn->send_mr->lkey;

    TEST_NZ(ibv_post_send(conn->qp, &wr, &bad_wr));
}

void send_ReadResponse(ReadResponse* response,Connection* conn)
{
    Client_Connection* client_conn = (Client_Connection*)conn->conn_info;

    struct ibv_send_wr wr, *bad_wr = NULL;
    struct ibv_sge sge;
    memset(&wr, 0, sizeof(wr));

    memcpy(client_conn->send_region,response,sizeof(ReadResponse));

    slot_info* info = (slot_info*)malloc(sizeof(slot_info));
    info->type=0;
    wr.wr_id = (uintptr_t)info;

    wr.opcode = IBV_WR_SEND;
    wr.sg_list = &sge;
    wr.num_sge = 1;
    wr.send_flags = IBV_SEND_SIGNALED;

    sge.addr = (uintptr_t)client_conn->send_region;
    sge.length = sizeof(ReadResponse);
    sge.lkey = client_conn->send_mr->lkey;

    TEST_NZ(ibv_post_send(conn->qp, &wr, &bad_wr));
}